package com.winner.rabbitmq.config;


import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


/**
 *
 * https://blog.csdn.net/u013871100/article/details/82982235
 *
 * @author wangxl
 * @date 2019-08-12
 */
@Configuration
@Slf4j
public class RabbitMqConfig {

//    @Autowired
//    private Environment env; 读取配置文件

    @Autowired
    private CachingConnectionFactory connectionFactory;

    @Autowired
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    /**
     * 单一消费者 消费者监听
     * @return
     */
    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainer(){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
//        https://www.jianshu.com/p/8614ff9ff0f1
//        //单台并发消费者数量
//        factory.setConcurrentConsumers(10);
//        //单台并发消费的最大消费者数量
//        factory.setMaxConcurrentConsumers(30);
//        //预取消费数量,unacked数量超过这个值broker将不会接收消息
//        factory.setPrefetchCount(5);
//        //有事务时处理的消息数
//        factory.setTxSize(1);
        factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
        return factory;
    }

    /**
     * 多个消费者
     * @return
     */
    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainer(){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factoryConfigurer.configure(factory,connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(20);
        factory.setPrefetchCount(20);
        return factory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(){
        connectionFactory.setPublisherConfirms(true); //Publisher Confirms and Returns
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        //通过Publisher Confirms and Returns机制，生产者可以判断消息是否发送到了exchange及queue
        /**
         * ConfirmCallback的回调说明:
         * 1.如果消息没有到exchange,则confirm回调,ack=false
         * 2.如果消息到达exchange,则confirm回调,ack=true
         */
        //消息发送到交换机确认机制
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if (ack) {
//                    String id = correlationData.getId();
                    log.info("消息发送exchange成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
                }else {
                    log.info("消息发送exchange失败:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
                }
            }
        });
       /**
        * ReturnCallback的回调说明:
        * 1. exchange到queue成功,则不回调 ReturnCallback
        * 2. exchange到queue失败,则回调  ReturnCallback(需设置mandatory=true,否则不回回调,消息就丢了)
        */
        //消息发送失败返回监听器
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message);
            }
        });
        return rabbitTemplate;
    }

    //    首先我们需要在上面的 RabbitmqConfig 类中创建消息模型：包括 Queue、Exchange、RoutingKey 绑定等 等的建立
    /** 消息交换机的名字*/
    public static final String EXCHANGE = "exchangeTest";
    /** 队列key1*/
    public static final String ROUTINGKEY1 = "queue_one_key1";
    /** 队列key2*/
    public static final String ROUTINGKEY2 = "queue_one_key2";

    @Bean
    public Queue firstQueue() {
        /**
         durable="true" 持久化 rabbitmq重启的时候不需要创建新的队列
         auto-delete 表示当所有消费客户端连接断开后（消息队列没有在使用时），是否自动删除队列 默认false不删除
         exclusive  表示该消息队列是否只在当前connection生效,默认是false
         */
        return new Queue("first-queue",true,false,false);
    }

    @Bean
    public Queue secondQueue() {
        return new Queue("second-queue",true,false,false);
    }

    /**
     * RabbitMQ提供了4种Exchange模式：  DirectExchange  FanoutExchange  TopicExchange HeadersExchange
     * direct (发布与订阅完全匹配)
     * fanout 广播 （没有路由键RouteKey的概念）
     * topic 主题，规则匹配
     *
     * direct交换器相对来说比较简单，匹配规则为：如果路由键匹配，消息就被投送到相关的队列
     * fanout交换器中没有路由键的概念，他会把消息发送到所有绑定在此交换器上面的队列中。
     * topic交换器你采用模糊匹配路由键的原则进行转发消息到队列中
     *
     * 交换器和队列的关系：
     * 交换器是通过 路由键 和 队列 绑定在一起的， 如果消息拥有的 路由键 跟队列 和 交换器 的路由建匹配， 那么消息
     * 就会被路由到该绑定的队列中
     * 也就是说， 消息队列的过程中， 消息首先会经过交换器， 接下来交换器在通过路由键值 匹配分发消息到具体的队列中。
     * 路由键 可以理解为 匹配的规则
     *
     * AMQP协议中的核心思想就是生产者和消费者隔离，生产者从不直接将消息发送给队列。
     * 生产者通常不知道是否一个消息会被发送到队列中，只是将消息发送到一个交换机。 先由Exchange来接收，然后Exchange按照特定的策略转发到Queue进行存储。
     * 同理，消费者也是如此。Exchange 就类似于一个交换机，转发各个消息分发到相应的队列中。 所有发送到Direct Exchange的消息被转发到RouteKey中指定的Queue
     *
     * 1.直接交换机
     */
    @Bean
    public DirectExchange directExchange(){
        //durable="true" rabbitmq重启的时候不需要创建新的交换机
        //当所有绑定队列都不在使用时，是否自动删除交换器 true表示删除  false表示不删除
        DirectExchange directExchange = new DirectExchange(RabbitMqConfig.EXCHANGE,true,false);
        return directExchange;
    }


    /**
     * 2.广播交换机
     * @return
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        FanoutExchange fanoutExchange = new FanoutExchange("fanoutExchange",true,false);
        return fanoutExchange;
    }

    /**
     * 3.主题交换机
     * @return
     */
    @Bean
    public TopicExchange topicExchange(){
        TopicExchange topicExchange = new TopicExchange("topicExchange",true,false);
        return topicExchange;
    }

    /**
     将消息队列1和交换机进行绑定
     */
    @Bean
    public Binding binding_one() {
        return BindingBuilder.bind(firstQueue()).to((directExchange())).with(RabbitMqConfig.ROUTINGKEY1);
    }

    /**
     * 将消息队列2和交换机进行绑定
     */
    @Bean
    public Binding binding_two() {
        return BindingBuilder.bind(firstQueue()).to((directExchange())).with(RabbitMqConfig.ROUTINGKEY2);
    }

}


